﻿#region Copyright

//  ======================================================
//      Copyright (c) 2011 Toulr All rights reserved.
//     
//      The use and distribution terms for this software are contained in the file
//      named license.txt, which can be found in the root of this distribution.
//      By using this software in any fashion, you are agreeing to be bound by the
//      terms of this license.
//     
//     You must not remove this notice, or any other, from this software.
//  ======================================================

#endregion

#region Using Namespaces

using System;
using System.Collections.Generic;
using System.Threading;
using Apache.NMS;
using Apache.NMS.ActiveMQ;
using Apache.NMS.ActiveMQ.Commands;
using Toulr.Util.Logging;

#endregion

namespace Toulr.Util.Net
{
    /// <summary>
    ///   ActiveMQ消息操作帮助类
    /// </summary>
    public class ActiveMq
    {
        private readonly IConnection _connection;
        private readonly IConnectionFactory _factory;
        private readonly ILogger _logger = LogHelper.LogFactoy.Create("ActiveMq");
        private bool _listenerStoped = false;

        public ActiveMq(String brokerUri, String clientId)
        {
            _factory = new ConnectionFactory(brokerUri);
            _connection = _factory.CreateConnection();
            _connection.ClientId = clientId;
        }

        public void Dispose()
        {
            try
            {
                _connection.Close();
            }
            catch (Exception ex)
            {
                _logger.Error(ex.Message, ex);
            }
        }

        /// <summary>
        ///   结束消息监听
        /// </summary>
        public void StopListener()
        {
            _listenerStoped = true;
        }

        /// <summary>
        ///   接收特定消息集合内的消消息
        /// </summary>
        public void StartListener(List<String> topics)
        {
            try
            {
                _connection.Start();

                //Create the Session
                using (ISession session = _connection.CreateSession())
                {
                    foreach (string topic in topics)
                    {
                        //Create the Consumer
                        IMessageConsumer consumer = session.CreateDurableConsumer(new ActiveMQTopic(topic),
                                                                                  _connection.ClientId, null, false);
                        consumer.Listener += new MessageListener(ConsumerListener);
                    }
                    while (!_listenerStoped) Thread.CurrentThread.Join(1000);
                }
                _connection.Stop();
                _connection.Close();
            }
            catch (Exception ex)
            {
                _logger.Error(ex.Message, ex);
            }
        }

        public void ConsumerListener(IMessage message)
        {
            var textMessage = (ITextMessage) message;
            _logger.Debug(textMessage.Text);
        }

        /// <summary>
        ///   向指定TOPIC发送消息
        /// </summary>
        /// <param name = "topic"></param>
        /// <param name = "messages"></param>
        public void SendMessage(String topic, List<String> messages)
        {
            using (ISession session = _connection.CreateSession())
            {
                IMessageProducer prod = session.CreateProducer(
                    new ActiveMQTopic(topic));
                foreach (string message in messages)
                {
                    ITextMessage msg = prod.CreateTextMessage();
                    msg.Text = message;
                    prod.Send(msg, MsgDeliveryMode.NonPersistent, MsgPriority.Normal, TimeSpan.MinValue);
                }
            }
        }
    }
}